//! 真正的单线程Redis架构
//! 
//! 完全避免锁竞争，所有操作在单个线程中串行执行
//! 参考Redis官方设计，追求极致性能
//! 支持不同启动模式（standalone/cluster），在standalone模式下禁用CLUSTER命令

use crate::storage::{HighPerformanceStorage, CommandResult};
use flume::Sender;
// use flume::Receiver;  // 已注释：未使用的导入
use std::sync::Arc;
use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
use std::thread;
use std::time::{/*Duration, */Instant};
use tracing::{info, warn/*, error*/};

/// Redis服务器运行模式
#[derive(Debug, Clone, PartialEq)]
pub enum RedisServerMode {
    /// 单机模式 - CLUSTER命令不可用
    Standalone,
    /// 集群模式 - CLUSTER命令可用
    Cluster,
}

/// Redis命令请求
#[derive(Debug)]
pub struct RedisCommand {
    pub command: String,
    pub args: Vec<String>,
    pub response_sender: flume::Sender<CommandResult>,
    pub timestamp: Instant,
}

impl RedisCommand {
    pub fn new(command: String, args: Vec<String>) -> (Self, flume::Receiver<CommandResult>) {
        let (tx, rx) = flume::unbounded();
        let cmd = Self {
            command,
            args,
            response_sender: tx,
            timestamp: Instant::now(),
        };
        (cmd, rx)
    }
}

/// 单线程Redis服务器统计
#[derive(Debug, Default)]
pub struct SingleThreadRedisStats {
    pub total_commands: AtomicU64,
    pub processed_commands: AtomicU64,
    pub failed_commands: AtomicU64,
    pub avg_processing_time_us: AtomicU64,
    pub current_qps: AtomicU64,
}

/// 真正的单线程Redis服务器
/// 
/// 设计原则：
/// 1. 所有命令在单个线程中串行执行
/// 2. 零锁竞争，零上下文切换
/// 3. 使用高性能无锁通道通信
/// 4. 极简设计，追求极致性能
/// 5. 支持不同运行模式，在standalone模式下禁用CLUSTER命令
pub struct SingleThreadRedisServer {
    command_sender: Sender<RedisCommand>,
    worker_handle: Option<thread::JoinHandle<()>>,
    running: Arc<AtomicBool>,
    stats: Arc<SingleThreadRedisStats>,
    /// 服务器运行模式
    mode: RedisServerMode,
}

impl Default for SingleThreadRedisServer {
    fn default() -> Self {
        Self::new()
    }
}

impl SingleThreadRedisServer {
    /// 创建新的单线程Redis服务器（默认standalone模式）
    pub fn new() -> Self {
        Self::new_with_mode(RedisServerMode::Standalone)
    }
    
    /// 创建指定模式的单线程Redis服务器
    pub fn new_with_mode(mode: RedisServerMode) -> Self {
        let (command_sender, _command_receiver) = flume::unbounded();
        let running = Arc::new(AtomicBool::new(false));
        let stats = Arc::new(SingleThreadRedisStats::default());

        info!("🏗️ 创建单线程Redis服务器，模式: {:?}", mode);

        Self {
            command_sender,
            worker_handle: None,
            running,
            stats,
            mode,
        }
    }

    /// 启动单线程Redis服务器
    pub fn start(&mut self) {
        if self.running.load(Ordering::Relaxed) {
            warn!("Single thread Redis server is already running");
            return;
        }

        self.running.store(true, Ordering::Relaxed);
        
        // 获取接收器用于工作线程
        let (_, _command_receiver) = flume::unbounded::<RedisCommand>();
        // 替换原有的接收器
        let (new_sender, new_receiver) = flume::unbounded();
        self.command_sender = new_sender;
        
        let running = self.running.clone();
        let stats = self.stats.clone();
        let mode = self.mode.clone();

        info!("🚀 启动单线程Redis服务器（零锁设计，模式: {:?}）", mode);

        let handle = thread::Builder::new()
            .name("redis-worker".to_string())
            .spawn(move || {
                Self::worker_loop(new_receiver, running, stats, mode);
            })
            .expect("Failed to spawn Redis worker thread");

        self.worker_handle = Some(handle);
        info!("✅ 单线程Redis服务器启动成功");
    }

    /// 停止服务器
    pub fn stop(&mut self) {
        if !self.running.load(Ordering::Relaxed) {
            return;
        }

        self.running.store(false, Ordering::Relaxed);

        if let Some(handle) = self.worker_handle.take() {
            let _ = handle.join();
        }

        info!("🛑 单线程Redis服务器已停止");
    }

    /// 执行Redis命令（异步接口）
    pub async fn execute_command(&self, command: String, args: Vec<String>) -> Result<CommandResult, String> {
        let (redis_cmd, response_receiver) = RedisCommand::new(command, args);
        
        // 发送命令到工作线程
        self.command_sender.send(redis_cmd)
            .map_err(|_| "Failed to send command to worker".to_string())?;

        self.stats.total_commands.fetch_add(1, Ordering::Relaxed);

        // 等待响应
        response_receiver.recv_async().await
            .map_err(|_| "Failed to receive response from worker".to_string())
    }

    /// 同步执行Redis命令
    pub fn execute_command_sync(&self, command: String, args: Vec<String>) -> Result<CommandResult, String> {
        let (redis_cmd, response_receiver) = RedisCommand::new(command, args);
        
        // 发送命令到工作线程
        self.command_sender.send(redis_cmd)
            .map_err(|_| "Failed to send command to worker".to_string())?;

        self.stats.total_commands.fetch_add(1, Ordering::Relaxed);

        // 等待响应
        response_receiver.recv()
            .map_err(|_| "Failed to receive response from worker".to_string())
    }

    /// 工作线程主循环
    fn worker_loop(
        command_receiver: flume::Receiver<RedisCommand>,
        running: Arc<AtomicBool>,
        stats: Arc<SingleThreadRedisStats>,
        mode: RedisServerMode,
    ) {
        // 创建存储引擎（单线程独占，无锁）
        let mut storage = HighPerformanceStorage::new();
        let mut processed_count = 0u64;
        let mut last_qps_update = Instant::now();
        let mut last_processed = 0u64;

        info!("🏃 Redis工作线程开始运行");

        while running.load(Ordering::Relaxed) {
            match command_receiver.recv() {
                Ok(command) => {

            let start_time = Instant::now();
            
            // 执行命令（单线程，无锁，极速）
            let result = Self::execute_redis_command(&mut storage, &command.command, command.args, &mode);
            
            // 发送响应
            let _ = command.response_sender.send(result);

            // 更新统计
            let processing_time = start_time.elapsed().as_micros() as u64;
            stats.avg_processing_time_us.store(processing_time, Ordering::Relaxed);
            stats.processed_commands.fetch_add(1, Ordering::Relaxed);
            
                    processed_count += 1;

                    // 每10万条命令更新一次QPS统计
                    if processed_count % 100_000 == 0 {
                        let elapsed = last_qps_update.elapsed();
                        let current_qps = (processed_count - last_processed) as f64 / elapsed.as_secs_f64();
                        stats.current_qps.store(current_qps as u64, Ordering::Relaxed);
                        
                        info!("🏃‍♂️ 已处理 {} 命令，当前QPS: {:.0}", processed_count, current_qps);
                        
                        last_qps_update = Instant::now();
                        last_processed = processed_count;
                    }
                }
                Err(_) => {
                    // 通道关闭，退出循环
                    break;
                }
            }
        }

        info!("🏁 Redis工作线程已停止，总共处理 {} 命令", processed_count);
    }

    /// 执行单个Redis命令（纯函数，极速）
    fn execute_redis_command(
        storage: &mut HighPerformanceStorage,
        command_name: &str,
        args: Vec<String>,
        mode: &RedisServerMode,
    ) -> CommandResult {
        match command_name.to_uppercase().as_str() {
            "PING" => CommandResult::String("PONG".to_string()),
            
            "SET" => {
                if args.len() != 2 {
                    return CommandResult::Error("SET requires exactly 2 arguments".to_string());
                }
                match storage.set(args[0].clone(), args[1].clone()) {
                    Ok(()) => CommandResult::Ok,
                    Err(e) => CommandResult::Error(e.to_string()),
                }
            }
            
            "GET" => {
                if args.len() != 1 {
                    return CommandResult::Error("GET requires exactly 1 argument".to_string());
                }
                match storage.get(&args[0]) {
                    Ok(Some(value)) => CommandResult::String(value),
                    Ok(None) => CommandResult::Null,
                    Err(e) => CommandResult::Error(e.to_string()),
                }
            }
            
            "EXISTS" => {
                if args.len() != 1 {
                    return CommandResult::Error("EXISTS requires exactly 1 argument".to_string());
                }
                let exists = storage.exists(&args[0]);
                CommandResult::Integer(if exists { 1 } else { 0 })
            }
            
            "DEL" => {
                if args.len() != 1 {
                    return CommandResult::Error("DEL requires exactly 1 argument".to_string());
                }
                let deleted = storage.del(&args[0]);
                CommandResult::Integer(if deleted { 1 } else { 0 })
            }
            
            "INCR" => {
                if args.len() != 1 {
                    return CommandResult::Error("INCR requires exactly 1 argument".to_string());
                }
                match storage.incr(&args[0]) {
                    Ok(value) => CommandResult::Integer(value),
                    Err(e) => CommandResult::Error(e.to_string()),
                }
            }
            
            "INCRBY" => {
                if args.len() != 2 {
                    return CommandResult::Error("INCRBY requires exactly 2 arguments".to_string());
                }
                match args[1].parse::<i64>() {
                    Ok(amount) => match storage.incr_by(&args[0], amount) {
                        Ok(value) => CommandResult::Integer(value),
                        Err(e) => CommandResult::Error(e.to_string()),
                    },
                    Err(_) => CommandResult::Error("Invalid number format".to_string()),
                }
            }
            
            "DECR" => {
                if args.len() != 1 {
                    return CommandResult::Error("DECR requires exactly 1 argument".to_string());
                }
                match storage.incr_by(&args[0], -1) {
                    Ok(value) => CommandResult::Integer(value),
                    Err(e) => CommandResult::Error(e.to_string()),
                }
            }
            
            "DECRBY" => {
                if args.len() != 2 {
                    return CommandResult::Error("DECRBY requires exactly 2 arguments".to_string());
                }
                match args[1].parse::<i64>() {
                    Ok(amount) => match storage.incr_by(&args[0], -amount) {
                        Ok(value) => CommandResult::Integer(value),
                        Err(e) => CommandResult::Error(e.to_string()),
                    },
                    Err(_) => CommandResult::Error("value is not an integer or out of range".to_string()),
                }
            }
            
            "HSET" => {
                if args.len() != 3 {
                    return CommandResult::Error("HSET requires exactly 3 arguments".to_string());
                }
                match storage.hset(&args[0], args[1].clone(), args[2].clone()) {
                    Ok(is_new) => CommandResult::Integer(if is_new { 1 } else { 0 }),
                    Err(e) => CommandResult::Error(e.to_string()),
                }
            }
            
            "HGET" => {
                if args.len() != 2 {
                    return CommandResult::Error("HGET requires exactly 2 arguments".to_string());
                }
                match storage.hget(&args[0], &args[1]) {
                    Ok(Some(value)) => CommandResult::String(value),
                    Ok(None) => CommandResult::Null,
                    Err(e) => CommandResult::Error(e.to_string()),
                }
            }
            
            "HDEL" => {
                if args.len() != 2 {
                    return CommandResult::Error("HDEL requires exactly 2 arguments".to_string());
                }
                match storage.hdel(&args[0], &args[1]) {
                    Ok(deleted) => CommandResult::Integer(if deleted { 1 } else { 0 }),
                    Err(e) => CommandResult::Error(e.to_string()),
                }
            }
            
            "FLUSHALL" => {
                storage.clear();
                CommandResult::Ok
            }
            
            "CLIENT" => {
                // CLIENT 命令简单处理
                if args.is_empty() {
                    CommandResult::Error("CLIENT requires subcommand".to_string())
                } else {
                    match args[0].to_uppercase().as_str() {
                        "SETINFO" => CommandResult::Ok,
                        "GETNAME" => CommandResult::Null,
                        "ID" => CommandResult::Integer(1),
                        "LIST" => CommandResult::String("".to_string()),
                        _ => CommandResult::Error(format!("Unknown CLIENT subcommand: {}", args[0]))
                    }
                }
            }
            
            "TYPE" => {
                if args.len() != 1 {
                    return CommandResult::Error("TYPE requires exactly 1 argument".to_string());
                }
                match storage.get_type(&args[0]) {
                    Some(type_name) => CommandResult::String(type_name),
                    None => CommandResult::String("none".to_string()),
                }
            }
            
            "EXPIRE" => {
                if args.len() != 2 {
                    return CommandResult::Error("EXPIRE requires exactly 2 arguments".to_string());
                }
                match args[1].parse::<u64>() {
                    Ok(seconds) => {
                        let result = storage.expire(&args[0], seconds);
                        CommandResult::Integer(if result { 1 } else { 0 })
                    },
                    Err(_) => CommandResult::Error("Invalid number format".to_string()),
                }
            }
            
            "TTL" => {
                if args.len() != 1 {
                    return CommandResult::Error("TTL requires exactly 1 argument".to_string());
                }
                let ttl = storage.ttl(&args[0]);
                CommandResult::Integer(ttl)
            }
            
            "PERSIST" => {
                if args.len() != 1 {
                    return CommandResult::Error("PERSIST requires exactly 1 argument".to_string());
                }
                let result = storage.persist(&args[0]);
                CommandResult::Integer(if result { 1 } else { 0 })
            }
            
            "CLUSTER" => {
                // 检查运行模式（使用您的思路：直接判断全局模式）
                match mode {
                    RedisServerMode::Standalone => {
                        CommandResult::Error("This instance has cluster support disabled".to_string())
                    }
                    RedisServerMode::Cluster => {
                        // 在集群模式下，处理CLUSTER命令
                        Self::execute_cluster_command(args)
                    }
                }
            }
            
            _ => CommandResult::Error(format!("Unknown command: {command_name}")),
        }
    }
    
    /// 处理CLUSTER命令（在集群模式下）
    fn execute_cluster_command(args: Vec<String>) -> CommandResult {
        if args.is_empty() {
            return CommandResult::Error("CLUSTER command requires subcommand".to_string());
        }
        
        let subcommand = args[0].to_uppercase();
        
        match subcommand.as_str() {
            "INFO" => {
                // 返回集群信息
                CommandResult::String(
                    "cluster_state:ok\r\n\
                     cluster_slots_assigned:16384\r\n\
                     cluster_slots_ok:16384\r\n\
                     cluster_slots_pfail:0\r\n\
                     cluster_slots_fail:0\r\n\
                     cluster_known_nodes:1\r\n\
                     cluster_size:1\r\n\
                     cluster_current_epoch:1\r\n\
                     cluster_my_epoch:1".to_string()
                )
            }
            "NODES" => {
                // 返回节点信息
                CommandResult::String(
                    "node1 127.0.0.1:7001@17001 master - 0 0 1 connected 0-16383".to_string()
                )
            }
            "SLOTS" => {
                // 返回槽位分配信息（使用RESP数组格式）
                CommandResult::Array(vec![
                    CommandResult::Array(vec![
                        CommandResult::Integer(0),
                        CommandResult::Integer(16383),
                        CommandResult::Array(vec![
                            CommandResult::String("127.0.0.1".to_string()),
                            CommandResult::Integer(7001),
                        ])
                    ])
                ])
            }
            "KEYSLOT" => {
                if args.len() != 2 {
                    return CommandResult::Error("CLUSTER KEYSLOT requires key".to_string());
                }
                let key = &args[1];
                // 简化的CRC16算法（实际中应该使用Redis标准CRC16）
                let mut crc: u16 = 0;
                for byte in key.bytes() {
                    crc = crc.wrapping_add(byte as u16);
                }
                let slot = crc % 16384;
                CommandResult::Integer(slot as i64)
            }
            "MYID" => {
                CommandResult::String("node1".to_string())
            }
            _ => CommandResult::Error(format!("Unknown CLUSTER subcommand: {}", subcommand))
        }
    }

    /// 获取服务器运行模式
    pub fn get_mode(&self) -> &RedisServerMode {
        &self.mode
    }
    
    /// 设置服务器运行模式（仅在服务器停止时可用）
    pub fn set_mode(&mut self, mode: RedisServerMode) -> Result<(), String> {
        if self.is_running() {
            return Err("Cannot change mode while server is running".to_string());
        }
        info!("🔄 更改Redis服务器模式: {:?} -> {:?}", self.mode, mode);
        self.mode = mode;
        Ok(())
    }

    /// 获取统计信息
    pub fn get_stats(&self) -> &SingleThreadRedisStats {
        &self.stats
    }

    /// 获取当前QPS
    pub fn get_current_qps(&self) -> u64 {
        self.stats.current_qps.load(Ordering::Relaxed)
    }

    /// 检查是否正在运行
    pub fn is_running(&self) -> bool {
        self.running.load(Ordering::Relaxed)
    }
}

impl Drop for SingleThreadRedisServer {
    fn drop(&mut self) {
        self.stop();
    }
}